import pymysql
import redis
from kafka import KafkaConsumer, KafkaProducer
import json
import time
from config.config import MYSQL_CONFIG, REDIS_CONFIG, KAFKA_CONFIG
import threading

# 全局变量用于跟踪数据库连接状态
_mysql_conn_lock = threading.Lock()
_redis_conn_lock = threading.Lock()
_mysql_conn = None
_redis_conn = None
_last_mysql_check = 0
_last_redis_check = 0
_mysql_check_interval = 60  # 每分钟检查一次

def get_mysql_conn(max_retries=3, retry_delay=2, force_new=False):
    """获取MySQL数据库连接，支持自动重连和连接池
    
    参数:
        max_retries: 最大重试次数
        retry_delay: 重试间隔(秒)
        force_new: 是否强制创建新连接
    """
    global _mysql_conn, _last_mysql_check
    
    # 检查是否需要创建新连接
    with _mysql_conn_lock:
        current_time = time.time()
        
        # 如果已有连接且未过期，直接返回
        if not force_new and _mysql_conn is not None:
            # 定期检查连接是否有效
            if current_time - _last_mysql_check >= _mysql_check_interval:
                if not _test_mysql_connection(_mysql_conn):
                    print("MySQL连接已断开，重新创建连接...")
                    try:
                        _mysql_conn.close()
                    except:
                        pass
                    _mysql_conn = None
                else:
                    _last_mysql_check = current_time
                    return _mysql_conn
            else:
                return _mysql_conn
        
        # 创建新连接
        retries = 0
        last_error = None
        
        while retries < max_retries:
            try:
                conn = pymysql.connect(
                    host=MYSQL_CONFIG['host'],
                    port=MYSQL_CONFIG['port'],
                    user=MYSQL_CONFIG['user'],
                    password=MYSQL_CONFIG['password'],
                    db=MYSQL_CONFIG['db'],
                    charset=MYSQL_CONFIG['charset'],
                    cursorclass=pymysql.cursors.DictCursor,
                    # 添加自动重连和连接超时配置
                    connect_timeout=10,
                    autocommit=True,
                    read_timeout=30,
                    write_timeout=30
                )
                
                # 测试连接是否有效
                with conn.cursor() as cursor:
                    cursor.execute("SELECT 1")
                    cursor.fetchone()
                
                print(f"MySQL数据库连接成功")
                _mysql_conn = conn
                _last_mysql_check = current_time
                return conn
            except Exception as e:
                last_error = e
                retries += 1
                print(f"MySQL连接尝试 {retries}/{max_retries} 失败: {e}")
                
                if retries < max_retries:
                    time.sleep(retry_delay)
        
        print(f"MySQL连接失败，已达到最大重试次数: {last_error}")
        return None

def get_redis_conn(max_retries=3, retry_delay=2, force_new=False):
    """获取Redis连接，支持自动重连和连接池
    
    参数:
        max_retries: 最大重试次数
        retry_delay: 重试间隔(秒)
        force_new: 是否强制创建新连接
    """
    global _redis_conn, _last_redis_check
    
    # 检查是否需要创建新连接
    with _redis_conn_lock:
        current_time = time.time()
        
        # 如果已有连接且未强制更新，直接返回
        if not force_new and _redis_conn is not None:
            # 定期检查连接是否有效
            if current_time - _last_redis_check >= _mysql_check_interval:
                try:
                    _redis_conn.ping()
                    _last_redis_check = current_time
                    return _redis_conn
                except:
                    print("Redis连接已断开，重新创建连接...")
                    _redis_conn = None
            else:
                return _redis_conn
        
        # 创建新连接
        retries = 0
        last_error = None
        
        while retries < max_retries:
            try:
                r = redis.Redis(
                    host=REDIS_CONFIG['host'],
                    port=REDIS_CONFIG['port'],
                    db=REDIS_CONFIG['db'],
                    password=REDIS_CONFIG['password'],
                    decode_responses=REDIS_CONFIG['decode_responses'],
                    socket_timeout=10,
                    socket_connect_timeout=10,
                    health_check_interval=30
                )
                
                # 测试连接是否有效
                r.ping()
                
                print(f"Redis连接成功")
                _redis_conn = r
                _last_redis_check = current_time
                return r
                
            except Exception as e:
                last_error = e
                retries += 1
                print(f"Redis连接尝试 {retries}/{max_retries} 失败: {e}")
                
                if retries < max_retries:
                    time.sleep(retry_delay)
        
        print(f"Redis连接失败，已达到最大重试次数: {last_error}")
        return None

def _test_mysql_connection(conn):
    """测试MySQL连接是否有效"""
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT 1")
            cursor.fetchone()
        return True
    except Exception:
        return False

def get_kafka_consumer():
    """获取Kafka消费者"""
    try:
        consumer = KafkaConsumer(
            KAFKA_CONFIG['user_behavior_topic'],
            bootstrap_servers=KAFKA_CONFIG['bootstrap_servers'],
            group_id=KAFKA_CONFIG['group_id'],
            auto_offset_reset='latest',
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )
        return consumer
    except Exception as e:
        print(f"Kafka消费者创建失败: {e}")
        return None

def get_kafka_producer():
    """获取Kafka生产者"""
    try:
        producer = KafkaProducer(
            bootstrap_servers=KAFKA_CONFIG['bootstrap_servers'],
            value_serializer=lambda m: json.dumps(m).encode('utf-8')
        )
        return producer
    except Exception as e:
        print(f"Kafka生产者创建失败: {e}")
        return None

def close_all_connections():
    """关闭所有数据库连接"""
    global _mysql_conn, _redis_conn
    
    with _mysql_conn_lock:
        if _mysql_conn:
            try:
                _mysql_conn.close()
                print("MySQL连接已关闭")
            except:
                pass
            _mysql_conn = None
    
    with _redis_conn_lock:
        if _redis_conn:
            try:
                _redis_conn.close()
                print("Redis连接已关闭")
            except:
                pass
            _redis_conn = None 